package streamAPI.operators;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.testng.annotations.Test;

@Slf4j
public class DemoWithIterate {
	@Test
	public void t1() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> source = env.socketTextStream("localhost", 4444);
		SingleOutputStreamOperator<Long> numbers = source.map(Long::parseLong);

		IterativeStream<Long> iterateStart = numbers.iterate();

		// 必须有一个更新模型   然后得到一个 iterationBody 就是更新之后的数据
		SingleOutputStreamOperator<Long> iterationBody = iterateStart.map(new MapFunction<Long, Long>() {
			@Override
			public Long map(Long value) throws Exception {
				log.warn("iterate ==>" + value);
				return value -= 1;
			}
		});

		// 只要满足下面的这个条件 就会形成一个回路 重新迭代 上面的更新模型
		SingleOutputStreamOperator<Long> feedback = iterationBody.filter(new FilterFunction<Long>() {
			@Override
			public boolean filter(Long value) throws Exception {
				return value > 0;
			}
		});

		// 传入迭代条件
		iterateStart.closeWith(feedback);

		// 不满足迭代条件最后输出
		iterationBody.filter(new FilterFunction<Long>() {
			@Override
			public boolean filter(Long val) throws Exception {
				return val <= 0;
			}
		}).print();

		env.execute();
	}
}
